package com.gome.han.bigdata.spark.core.wordcount.wcount

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author Hanpeng
 * @date 2021/1/8 16:53
 * @description: 第一个wordcount 程序,也是一个driver program
 */
object WordCount01 {
  def main(args: Array[String]): Unit = {
    //01 app创建和spark 框架的连接(我们的目的时让spark 框架运行我们的程序)
    val sparkConf: SparkConf = new SparkConf();
    sparkConf.setAppName("WordCount01");
    //设置连接master的url
    //    sparkConf.setMaster("local[*]");
    //这种方式可以直接使用spark集群资源
    sparkConf.setMaster("spark://192.168.85.194:7077")
      //这个包指的是这个项目对应的jar
      .setJars(List("C:\\devSoft\\java\\workspacetianqiong\\bigdata\\spark\\target\\spark.jar"));
    //                   spark://master:7077
    val sc = new SparkContext(sparkConf); //可以run 一下 如果不报错 说明环境搭建正确
    sc.setLogLevel("ERROR")
    //02 处理业务
    //读取每行数据
    //    val lines:RDD[String] = sc.textFile("in");
    val array1 = Array("hello", "hello", "love", "you").toSeq;
    val parallelize = sc.parallelize(array1)
    val words = parallelize.flatMap(x=> {
      println(x) //这就是一个个的单词了
      //然后再split？？是这么的，这个用array 我是为了避免读取文件 直接内存创建RDD
      //免得出现路径问题
      //如果是内存创建的RDD 这个没有意义 但是不影响
      //但是如果是文件 读取的是 han I happy这种数据 split就有意义了
      val strings: Array[String] = x.split(" ")
      strings
    })
    //将每行数据进行拆分成一个个单词  也叫扁平化处理  将每个元素拆分成一个个单词
    //这样元素的单位是个word

    //    val wordGroup: RDD[(String, Iterable[String])] = parallelize.groupBy(_.toString)
    //将单词相同的单词进行分组 便于统计  (hello, hello, hello), (world, world)
    val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word);

    //对数据每个元素进行转换
    val wordToCount = wordGroup.map({
      case (word, list) => {(word, list.size)}
    });
    wordToCount.collect().foreach(println(_))
    //    val wordToCount1 = wordGroup.map(wordhan => {
    //      (wordhan._1, wordhan._2.size)
    //    });(println(_))
    /*map(x.match{case(x._1,x._2) => 后面是一样的}*/
    //    val l = parallelize.map((_, 1)).reduceByKey(_+_)
    //    wordToCount1.foreach(println(_))
    //    val wordToCount3 = map.map((han => {
    //      han match {
    //        case (x, y) => {
    //          (x, y.size)
    //        }
    //      }
    //    }));
    //这个执行的操作是  先将han赋值给(x, y)  看是否能这样赋值，如果成功(因为han是Tuple),则 看是否相等(显然相等) 执行操作
    //    val wordToCount3 = wordGroup.map(han =>
    //      han match {
    //        case (x, y) => (x, y.size)
    //      }
    //    );

    //打印结果
    //    val array = wordToCount3.collect()
    ////        array.foreach(println)
    //    val l = lines.count()
    //    wordToCount.collect().foreach(println(_))


    //03 关闭连接和spark 框架的连接
    sc.stop();

    //groupRDD.red

  }
}




